#include "message_db_wrapper.h"

#include "base/format_time.h"
#include <boost/algorithm/string.hpp>
#include <boost/typeof/typeof.hpp>

namespace imserver {

const int MessageDBWrapper::MESSAGE_P2P_INDEX_DB_MOD = 10;
const int MessageDBWrapper::MESSAGE_GROUP_INDEX_DB_MOD = 10;
const int MessageDBWrapper::MESSAGE_ENTITY_DB_MOD = 20;

bool MessageDBWrapper::select_tb_message_id_seq(int64_t& seq) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::select_tb_message_id_seq --> get db conn failed.");
    return false;
  }

  std::ostringstream os;
  os << "SELECT msg_id FROM tb_message_id_seq ORDER BY msg_id DESC LIMIT 1";
  mysqlpp::Query query(cp->query(os.str()));
  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while (mysqlpp::Row row = ret.fetch_row()) {
        seq = (int64_t)row["msg_id"];
      }
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::select_tb_message_id_seq --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }
  return true;
}

bool MessageDBWrapper::update_tb_message_id_seq() {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::update_tb_message_id_seq --> get db conn failed.");
    return false;
  }

  std::ostringstream os;
  os << "UPDATE tb_message_id_seq SET msg_id = msg_id + 1";
  mysqlpp::Query query(cp->query(os.str()));
  try {
    if (!query.exec()) {
      LOG_ERROR("MessageDBWrapper::update_tb_message_id_seq --> query return false. query_str:" << query.str());
      return false;
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::update_tb_message_id_seq --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }
  return true;
}

bool MessageDBWrapper::select_tb_dialogue_id_seq(int64_t& seq) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::select_tb_dialogue_id_seq --> get db conn failed.");
    return false;
  }

  std::ostringstream os;
  os << "SELECT dialogue_id FROM tb_dialogue_id_seq ORDER BY dialogue_id DESC LIMIT 1";
  mysqlpp::Query query(cp->query(os.str()));
  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while (mysqlpp::Row row = ret.fetch_row()) {
        seq = (int64_t)row["dialogue_id"];
      }
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::select_tb_dialogue_id_seq --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }

  return true;
}

bool MessageDBWrapper::update_tb_dialogue_id_seq() {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::update_tb_dialogue_id_seq --> get db conn failed.");
    return false;
  }

  std::ostringstream os;
  os << "UPDATE tb_dialogue_id_seq SET dialogue_id = dialogue_id + 1";
  mysqlpp::Query query(cp->query(os.str()));
  try {
    if (!query.exec()) {
      LOG_ERROR("MessageDBWrapper::update_tb_dialogue_id_seq --> query return false. query_str:" << query.str());
      return false;
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::update_tb_dialogue_id_seq --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }
  return true;
}

bool MessageDBWrapper::select_tb_dialogue(int64_t uid1, int64_t uid2, int64_t& dialogue_id) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::select_tb_dialogue --> get db conn failed. uid1:" << uid1 << " uid2:" << uid2);
    return false;
  }

  std::ostringstream os;
  os << "SELECT dialogue_id FROM tb_dialogue WHERE uid1 = " << (uid1 < uid2 ? uid1 : uid2) <<
      " AND uid2 = " << (uid1 > uid2 ? uid1 : uid2) << " LIMIT 1";
  mysqlpp::Query query(cp->query(os.str()));
  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while (mysqlpp::Row row = ret.fetch_row()) {
        dialogue_id = (int64_t)row["dialogue_id"];
      }
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::select_tb_dialogue --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }
  return true;
}

bool MessageDBWrapper::select_tb_dialogue(int64_t uid1, int64_t last_msg_id, std::set<int64_t>& dialogue_ids) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::select_tb_dialogue --> get db conn failed. uid1:" << uid1 << " last_msg_id:" <<
        last_msg_id);
    return false;
  }

  std::ostringstream os;
  os << "SELECT dialogue_id FROM tb_dialogue WHERE (uid1 = " << uid1 << " OR uid2 = " << uid1 << ") " <<
      " AND last_msg_id > " << last_msg_id;
  mysqlpp::Query query(cp->query(os.str()));
  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while (mysqlpp::Row row = ret.fetch_row()) {
        dialogue_ids.insert((int64_t)row["dialogue_id"]);
      }
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::select_tb_dialogue --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }
  return true;
}

bool MessageDBWrapper::update_tb_dialogue(int64_t dialogue_id, int64_t uid1, int64_t uid2, int64_t last_msg_id) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::update_tb_dialogue --> get db conn failed. dialogue_id:" << dialogue_id << 
        " last_msg_id:" << last_msg_id);
    return false;
  }

  std::ostringstream os;
  os << "INSERT INTO tb_dialogue (dialogue_id, uid1, uid2, last_msg_id) VALUES (" <<
      dialogue_id << "," <<
      (uid1 < uid2 ? uid1 : uid2) << "," <<
      (uid1 > uid2 ? uid1 : uid2) << "," <<
      last_msg_id << ") ON DUPLICATE KEY UPDATE last_msg_id = " <<
      last_msg_id;
  mysqlpp::Query query(cp->query(os.str()));
  try {
    if (!query.exec()) {
      LOG_ERROR("MessageDBWrapper::update_tb_dialogue --> query return false. query_str:" << query.str());
      return false;
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::update_tb_dialogue --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }
  return true;
}

bool MessageDBWrapper::insert_tb_message_entity(const ImMessage& message) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::insert_tb_message_entity --> get db conn failed. msg_id:" << message.id);
    return false;
  }

  std::string body(message.body);
  boost::algorithm::replace_all(body, "\\", "\\\\");
  boost::algorithm::replace_all(body, "\"", "\\\"");
  std::ostringstream os;
  os << "INSERT INTO tb_message_entity_" << message.dialogue_id % MESSAGE_ENTITY_DB_MOD <<
      " (msg_id, type, sender, receiver, create_date, body, group_id, dialogue_id, topic_id) VALUES (" <<
      message.id << "," <<
      message.type << "," <<
      message.sender << "," <<
      message.receiver << ",\"" <<
      base::FriendlyTime((time_t)message.timestamp) << "\",\"" <<
      body << "\"," <<
      message.group_id << "," <<
      message.dialogue_id << "," <<
      message.topic_id << ")";
  mysqlpp::Query query(cp->query(os.str()));
  try {
    if (!query.exec()) {
      LOG_ERROR("MessageDBWrapper::insert_tb_message_entity --> query return false. query_str:" << query.str());
      return false;
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::insert_tb_message_entity --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }
  LOG_INFO("MessageDBWrapper::insert_tb_message_entity --> ok. msg_id:" << message.id);
  return true;
}

bool MessageDBWrapper::select_tb_message_entity(int64_t uid, const std::map<int64_t, int64_t>& msgid_dialogid,
    std::map<int64_t, ImMessage>& res) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::select_tb_message_entity --> get db conn failed. uid:" << uid);
    return false;
  }

  std::vector<std::set<int64_t> > msgs;
  for (int i =0; i < MESSAGE_ENTITY_DB_MOD; ++i) {
    msgs.push_back(std::set<int64_t>());
  }
  BOOST_AUTO(it1, msgid_dialogid.begin());
  for (; it1 != msgid_dialogid.end(); ++it1) {
    msgs[(it1->second) % MESSAGE_ENTITY_DB_MOD].insert(it1->first);
  }

  mysqlpp::Query query(&(*cp));
  for (int i = 0; i < MESSAGE_ENTITY_DB_MOD; ++i) {
    if (msgs[i].empty()) {
      continue;
    }
    query << "SELECT msg_id, type, sender, receiver, body, create_date, group_id, dialogue_id FROM" <<
        " tb_message_entity_" << (i % MESSAGE_ENTITY_DB_MOD) <<
        " WHERE msg_id in (";
    BOOST_AUTO(it2, msgs[i].begin());
    for (; it2 != msgs[i].end(); ++it2) {
      if (it2 == msgs[i].begin()) {
        query << *it2;
      } else {
        query << "," << *it2;
      }
    }
    query << ")";
    try {
      mysqlpp::UseQueryResult ret = query.use();
      if (ret) {
        while (mysqlpp::Row row = ret.fetch_row()) {
          ImMessage msg;
          msg.id = (int64_t)row["msg_id"];
          msg.type = (int32_t)row["type"];
          msg.sender = (int64_t)row["sender"];
          msg.receiver = (int64_t)row["receiver"];
          msg.body = (std::string)row["body"];
          msg.timestamp = base::ToTimestamp(((std::string)row["create_date"]));
          msg.group_id = (int64_t)row["group_id"];
          msg.dialogue_id = (int64_t)row["dialogue_id"];
          res.insert(std::make_pair(msg.id, msg));
        }
      }
    } catch (mysqlpp::Exception &e) {
      LOG_ERROR("MessageDBWrapper::select_tb_message_entity --> query_str:" << query.str() << " error:" << e.what());
    }
    query.reset();
  }
  LOG_INFO("MessageDBWrapper::select_tb_message_entity --> query ok. uid:" << uid << " msg_ids_cout:" << msgid_dialogid.size() <<
      " res_cout:" << res.size());
  return true;
}

bool MessageDBWrapper::insert_tb_p2p_message_index(const ImMessage& message) {
  if (message.type != ImMessageType::IM_MESSAGE_P2P &&
      message.type != ImMessageType::IM_MESSAGE_RESUME &&
      message.type != ImMessageType::IM_MESSAGE_RESUME_LIKE &&
      message.type != ImMessageType::IM_MESSAGE_RESUME_IGNORE &&
      message.type != ImMessageType::IM_MESSAGE_JOB_INVITATION) {
    LOG_ERROR("MessageDBWrapper::insert_tb_p2p_message_index --> wrong message type. msg_id:" << message.id <<
        " type:" << message.type);
    return false;
  }
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::insert_tb_p2p_message_index --> get db conn failed. msg_id:" << message.id);
    return false;
  }

  std::ostringstream os;
  os << "INSERT INTO tb_p2p_message_index_" << message.dialogue_id % MESSAGE_P2P_INDEX_DB_MOD <<
      "(msg_id, type, sender, receiver, dialogue_id, topic_id, read_status, receive_status) values (" <<
      message.id << "," <<
      message.type << "," <<
      message.sender << "," <<
      message.receiver << "," <<
      message.dialogue_id << "," <<
      message.topic_id << ",0,0)";
  mysqlpp::Query query(cp->query(os.str()));
  try {
    if (!query.exec()) {
      LOG_ERROR("MessageDBWrapper::insert_tb_p2p_message_index --> query return false. query_str:" << query.str());
      return false;
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::insert_tb_p2p_message_index --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }
  LOG_INFO("MessageDBWrapper::insert_tb_p2p_message_entity --> query ok. dialogue_id:" << message.dialogue_id <<
      " msg_id:" << message.id);
  return true;
}

bool MessageDBWrapper::select_tb_p2p_message_index_u(int64_t uid, const std::set<int64_t>& dialogue_ids, 
    uint32_t unset_flag, std::map<int64_t, std::pair<int32_t, int64_t> >& res) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::select_tb_p2p_message_index_u --> get db conn failed. uid:" << uid);
    return false;
  }

  std::vector<std::set<int64_t> > dialogues;
  for (int i = 0; i < MESSAGE_P2P_INDEX_DB_MOD; ++i) {
    dialogues.push_back(std::set<int64_t>());
  }
  BOOST_AUTO(it1, dialogue_ids.begin());
  for (; it1 != dialogue_ids.end(); ++it1) {
    dialogues[(*it1)%MESSAGE_P2P_INDEX_DB_MOD].insert(*it1);
  }

  mysqlpp::Query query(&(*cp));
  for (int i =0; i < MESSAGE_P2P_INDEX_DB_MOD; ++i) {
    if (dialogues[i].empty()) {
      continue;
    }
    query << "SELECT dialogue_id, COUNT(1) as c, MAX(msg_id) as max_msg_id FROM tb_p2p_message_index_" << i <<
        " WHERE dialogue_id in (";
    BOOST_AUTO(it2, dialogues[i].begin());
    for (; it2 != dialogues[i].end(); ++it2) {
      if (it2 == dialogues[i].begin()) {
        query << *it2;
      } else {
        query << "," << *it2;
      }
    }
    query << ")";
    if (unset_flag & ImMessageFlag::IM_MESSAGE_RECEIVED) {
      query << " AND receive_status = 0";
    }
    if (unset_flag & ImMessageFlag::IM_MESSAGE_READ) {
      query << " AND read_status = 0";
    }
    query << " AND receiver = " << uid <<
        " AND type IN (" << ImMessageType::IM_MESSAGE_P2P << "," << ImMessageType::IM_MESSAGE_RESUME << 
        "," << ImMessageType::IM_MESSAGE_RESUME_LIKE << "," << ImMessageType::IM_MESSAGE_RESUME_IGNORE <<
        "," << ImMessageType::IM_MESSAGE_JOB_INVITATION << ")" <<
        " GROUP BY dialogue_id";
    try {
      mysqlpp::UseQueryResult ret = query.use();
      if (ret) {
        while (mysqlpp::Row row = ret.fetch_row()) {
          if ((int32_t)row["c"] <= 0) {
            continue;
          }
          res.insert(std::make_pair(
              (int64_t)row["dialogue_id"],
              std::make_pair((int32_t)row["c"], (int64_t)row["max_msg_id"])));
        }
      }
    } catch (mysqlpp::Exception &e) {
      LOG_ERROR("MessageDBWrapper::select_tb_p2p_message_index_u --> query_str:" << query.str() << " error:" << e.what());
    }
    query.reset();
  }
  LOG_INFO("MessageDBWrapper::select_tb_p2p_message_index_u --> uid:" << uid << " res_count:" << res.size());
  return true;
}

bool MessageDBWrapper::select_tb_p2p_message_index_o(int64_t uid, int64_t dialogue_id, int64_t min_msg_id, int32_t limit,
    std::set<int64_t>& msgid_res) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::select_tb_p2p_message_index_o --> get db conn failed. uid:" << uid << " dialogue_id:" <<
        dialogue_id);
    return false;
  }

  std::set<int64_t> res;
  std::ostringstream os;
  os << "SELECT msg_id FROM tb_p2p_message_index_" << dialogue_id % MESSAGE_P2P_INDEX_DB_MOD <<
      " WHERE dialogue_id = " << dialogue_id << " AND receiver = " << uid;
  if (min_msg_id > 0) {
    os << " AND msg_id < " << min_msg_id;
  }
  os << " AND type IN (" << ImMessageType::IM_MESSAGE_P2P << "," << ImMessageType::IM_MESSAGE_RESUME << 
      "," << ImMessageType::IM_MESSAGE_RESUME_LIKE << "," << ImMessageType::IM_MESSAGE_RESUME_IGNORE <<
      "," << ImMessageType::IM_MESSAGE_JOB_INVITATION << ")" <<
      " ORDER BY msg_id DESC LIMIT " << limit;
  mysqlpp::Query query(cp->query(os.str()));
  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while (mysqlpp::Row row = ret.fetch_row()) {
        res.insert((int64_t)row["msg_id"]);
      }
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::select_tb_p2p_message_index_o --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }

  int64_t msg_id_t = 0L;
  if (res.size() >= (size_t)limit) {
    msg_id_t = *(res.begin());
  }

  query.reset();
  query << "SELECT msg_id FROM tb_p2p_message_index_" << dialogue_id % MESSAGE_P2P_INDEX_DB_MOD <<
      " WHERE dialogue_id = " << dialogue_id << " AND sender = " << uid << " AND msg_id > " << msg_id_t;
  if (min_msg_id > 0) {
    query << " AND msg_id < " << min_msg_id;
  }
  
  query << " AND type IN (" << ImMessageType::IM_MESSAGE_P2P << "," << ImMessageType::IM_MESSAGE_JOB_INVITATION << ")" <<
      " ORDER BY msg_id DESC LIMIT " << limit;
  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while (mysqlpp::Row row = ret.fetch_row()) {
        res.insert((int64_t)row["msg_id"]);
      }
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::select_tb_p2p_message_index_o --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }

  int i = 0;
  for (std::set<int64_t>::reverse_iterator it = res.rbegin(); i < 10 && it != res.rend(); ++it, ++i) {
    msgid_res.insert(*it);
  }

  LOG_INFO("MessageDBWrapper::select_tb_p2p_message_index_o --> uid:" << uid << " dialogue_id:" << dialogue_id <<
      " msgid_res_count:" << msgid_res.size());
  return true;
}

bool MessageDBWrapper::select_tb_p2p_message_index_n(int64_t uid, int64_t dialogue_id, int64_t max_msg_id, int32_t limit,
    std::set<int64_t>& msgid_res) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::select_tb_p2p_message_index_n --> get db conn failed. uid:" << uid << " dialogue_id:" <<
        dialogue_id);
    return false;
  }

  std::set<int64_t> res;
  std::ostringstream os;
  os << "SELECT msg_id FROM tb_p2p_message_index_" << dialogue_id % MESSAGE_P2P_INDEX_DB_MOD <<
      " WHERE dialogue_id = " << dialogue_id << " AND receiver = " << uid;
  if (max_msg_id > 0) {
    os << " AND msg_id > " << max_msg_id;
  }
  os << " AND type IN (" << ImMessageType::IM_MESSAGE_P2P << "," << ImMessageType::IM_MESSAGE_RESUME << 
      "," << ImMessageType::IM_MESSAGE_RESUME_LIKE << "," << ImMessageType::IM_MESSAGE_RESUME_IGNORE <<
      "," << ImMessageType::IM_MESSAGE_JOB_INVITATION << ")" <<
      " ORDER BY msg_id DESC LIMIT " << limit;
  mysqlpp::Query query(cp->query(os.str()));
  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while (mysqlpp::Row row = ret.fetch_row()) {
        res.insert((int64_t)row["msg_id"]);
      }
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::select_tb_p2p_message_index_n --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }

  int64_t msg_id_t = max_msg_id;
  if (res.size() >= (size_t)limit) {
    msg_id_t = *(res.begin());
  }

  query.reset();
  query << "SELECT msg_id FROM tb_p2p_message_index_" << dialogue_id % MESSAGE_P2P_INDEX_DB_MOD <<
      " WHERE dialogue_id = " << dialogue_id << " AND sender = " << uid;
  if (msg_id_t > 0) {
    query << " AND msg_id > " << msg_id_t;
  }
  query << " AND type IN (" << ImMessageType::IM_MESSAGE_P2P << "," << ImMessageType::IM_MESSAGE_JOB_INVITATION << ")" <<
      " ORDER BY msg_id DESC LIMIT " << limit;
  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while (mysqlpp::Row row = ret.fetch_row()) {
        res.insert((int64_t)row["msg_id"]);
      }
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::select_tb_p2p_message_index_o --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }

  int i = 0;
  for (std::set<int64_t>::reverse_iterator it = res.rbegin(); i < 10 && it != res.rend(); ++it, ++i) {
    msgid_res.insert(*it);
  }

  LOG_INFO("MessageDBWrapper::select_tb_p2p_message_index_n --> uid:" << uid << " dialogue_id:" << dialogue_id <<
      " msgid_res_count:" << msgid_res.size());
  return true;
}

bool MessageDBWrapper::update_tb_p2p_message_index(int64_t uid, int64_t dialogue_id, int64_t max_msgid,
    uint32_t set_flag) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::update_tb_p2p_message_index --> get db conn failed. uid:" << uid << " dialogue_id:" <<
        dialogue_id << " max_msgid:" << max_msgid << " set_flag:" << set_flag);
    return false;
  }
  if (max_msgid <= 0) {
    max_msgid = (std::numeric_limits<int64_t>::max)();
  }

  int64_t msgid_m = 0;
  std::ostringstream os;
  os << "SELECT MAX(msg_id) AS max_msgid FROM tb_p2p_message_index_" << dialogue_id % MESSAGE_P2P_INDEX_DB_MOD <<
        " WHERE dialogue_id = " << dialogue_id <<
        " AND type IN (" << ImMessageType::IM_MESSAGE_P2P << "," << ImMessageType::IM_MESSAGE_RESUME <<
        "," << ImMessageType::IM_MESSAGE_RESUME_LIKE << "," << ImMessageType::IM_MESSAGE_RESUME_IGNORE <<
        "," << ImMessageType::IM_MESSAGE_JOB_INVITATION << ")";
  if (set_flag & ImMessageFlag::IM_MESSAGE_RECEIVED) {
    os << " AND receive_status = 1";
  }
  if (set_flag & ImMessageFlag::IM_MESSAGE_READ) {
    os << " AND read_status = 1";
  }
  os << " AND receiver = " << uid << " LIMIT 1";
  mysqlpp::Query query(cp->query(os.str()));
  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while (mysqlpp::Row row = ret.fetch_row()) {
        if (row["max_msgid"] && row["max_msgid"] != "NULL") {
          msgid_m = (int64_t)row["max_msgid"];
        }
      }
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::update_tb_p2p_message_index --> query_str:" << os.str() << " error:" << e.what());
  }
  if (msgid_m >= max_msgid) {
    return true;
  }
  query.reset();
  query << "UPDATE tb_p2p_message_index_" << dialogue_id % MESSAGE_P2P_INDEX_DB_MOD;
  if (set_flag & ImMessageFlag::IM_MESSAGE_READ) {
    query << " SET read_status = 1, receive_status = 1";
  } else if (set_flag & ImMessageFlag::IM_MESSAGE_RECEIVED) {
    query << " SET receive_status = 1";
  }
  query << " WHERE msg_id > " << msgid_m << " AND msg_id <=" << max_msgid <<
      " AND type IN (" << ImMessageType::IM_MESSAGE_P2P << "," << ImMessageType::IM_MESSAGE_RESUME << 
      "," << ImMessageType::IM_MESSAGE_RESUME_LIKE << "," << ImMessageType::IM_MESSAGE_RESUME_IGNORE <<
      "," << ImMessageType::IM_MESSAGE_JOB_INVITATION << ")" <<
      " AND dialogue_id = " << dialogue_id <<
      " AND receiver = " << uid;
  try {
    if (!query.exec()) {
      LOG_ERROR("MessageDBWrapper::update_tb_p2p_message_index --> query return false. query_str:" << query.str());
      return false;
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::update_tb_p2p_message_index --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }

  LOG_INFO("MessageDBWrapper::update_tb_p2p_message_index --> ok. uid:" << uid << " dialogue_id:" << dialogue_id <<
      " max_msgid:" << max_msgid << " set_flag:" << set_flag);
  return true;
}

bool MessageDBWrapper::insert_tb_group_message_index(const ImMessage& message) {
  if (message.type != ImMessageType::IM_MESSAGE_GROUP &&
      message.type != ImMessageType::IM_MESSAGE_GROUP_JOIN &&
      message.type != ImMessageType::IM_MESSAGE_GROUP_RESUME_LIKE &&
      message.type != ImMessageType::IM_MESSAGE_GROUP_JOB_INVITATION) {
    LOG_ERROR("MessageDBWrapper::insert_tb_group_message_index --> message.type not IM_MESSAGE_GROUP. type:" <<
        message.type << " msg_id:" << message.id);
    return false;
  }
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::insert_tb_group_message_index --> get db conn failed. msg_id:" << message.id);
    return false;
  }

  std::ostringstream os;
  os << "INSERT INTO tb_group_message_index_" << message.group_id % MESSAGE_GROUP_INDEX_DB_MOD <<
      "(msg_id, type, sender, group_id) values (" <<
      message.id << "," <<
      message.type << "," <<
      message.sender << "," <<
      message.group_id << ")";
  mysqlpp::Query query(cp->query(os.str()));
  try {
    if (!query.exec()) { 
      LOG_ERROR("MessageDBWrapper::insert_tb_group_message_index --> query return false. query_str:" << query.str());
      return false;
    }
  } catch (mysqlpp::Exception &e) { 
    LOG_ERROR("MessageDBWrapper::insert_tb_group_message_index --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }

  LOG_INFO("MessageDBWrapper::insert_tb_group_message_index --> query ok. msg_id:" << message.id << " group_id:" <<
      message.group_id << " sender:" << message.sender);
  return true;
}

bool MessageDBWrapper::select_tb_group_message_index_u(int64_t uid, int64_t group_id, int64_t last_read_msgid,
    int32_t& unread_count, int64_t& max_msgid) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::select_tb_group_message_index_u --> get db conn failed. group_id:" << group_id <<
        " uid:" << uid);
    return false;
  }

  std::ostringstream os;
  os << "SELECT COUNT(1) as c, MAX(msg_id) as max_msgid FROM tb_group_message_index_" <<
      group_id % MESSAGE_GROUP_INDEX_DB_MOD <<
      " WHERE group_id = " << group_id <<
      " AND msg_id > " << last_read_msgid <<
      " AND sender != " << uid <<
      " AND type IN (" << ImMessageType::IM_MESSAGE_GROUP << "," <<
      ImMessageType::IM_MESSAGE_GROUP_JOIN << "," << ImMessageType::IM_MESSAGE_GROUP_RESUME_LIKE <<
      "," << ImMessageType::IM_MESSAGE_GROUP_JOB_INVITATION << ")" << 
      " limit 1";
  mysqlpp::Query query(cp->query(os.str()));
  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while (mysqlpp::Row row = ret.fetch_row()) {
        if ((int32_t)row["c"] <= 0) {
          continue;
        }
        unread_count = (int32_t)row["c"];
        max_msgid = (int64_t)row["max_msgid"];
      }
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::select_tb_group_message_index_u --> query_str:" << query.str() << " error:" <<
        e.what());
    return false;
  }

  LOG_INFO("MessageDBWrapper::select_tb_group_message_index_u --> query ok. group_id:" << group_id << " uid:" <<
      uid << " last_read_msgid:" << last_read_msgid << " unread_count:" << unread_count << " max_msgid:" << max_msgid);
  return true;
}

bool MessageDBWrapper::select_tb_group_message_index_o(int64_t uid, int64_t group_id, int64_t min_msg_id, int32_t limit,
    std::set<int64_t>& msgid_res) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::select_tb_group_message_index_o --> get db conn failed. uid:" << uid << " group_id:" <<
        group_id << " min_msg_id:" << min_msg_id << " limit:" << limit);
    return false;
  }

  std::ostringstream os;
  os << "SELECT msg_id FROM tb_group_message_index_" << group_id % MESSAGE_GROUP_INDEX_DB_MOD <<
      " WHERE group_id = " << group_id <<
      " AND type IN (" << ImMessageType::IM_MESSAGE_GROUP << "," << ImMessageType::IM_MESSAGE_GROUP_JOIN << "," <<
      ImMessageType::IM_MESSAGE_GROUP_RESUME_LIKE << "," << ImMessageType::IM_MESSAGE_GROUP_JOB_INVITATION << ")";
  if (min_msg_id > 0) {
    os << " AND msg_id < " << min_msg_id;
  }
  os << " ORDER BY msg_id desc LIMIT " << limit;
  mysqlpp::Query query(cp->query(os.str()));
  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while (mysqlpp::Row row = ret.fetch_row()) {
        msgid_res.insert((int64_t)row["msg_id"]);
      }
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::select_tb_group_message_index_o --> query_str:" << query.str() << " error:" <<
        e.what());
    return false;
  }

  LOG_INFO("MessageDBWrapper::select_tb_group_message_index_o --> ok. uid:" << uid << " group_id:" << group_id <<
      " min_msg_id:" << min_msg_id << " limit:" << limit << " msgid_res_count:" << msgid_res.size());
  return true;
}

bool MessageDBWrapper::select_tb_group_message_index_n(int64_t uid, int64_t group_id, int64_t max_msg_id, int32_t limit,
    std::set<int64_t>& msgid_res) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::select_tb_group_message_index_n --> get db conn failed. uid:" << uid << " group_id:" <<
        group_id << " max_msg_id:" << max_msg_id << " limit:" << limit);
    return false;
  }

  std::ostringstream os;
  os << "SELECT msg_id FROM tb_group_message_index_" << group_id % MESSAGE_GROUP_INDEX_DB_MOD <<
      " WHERE group_id = " << group_id <<
      " AND type IN (" << ImMessageType::IM_MESSAGE_GROUP << "," << ImMessageType::IM_MESSAGE_GROUP_JOIN << "," <<
      ImMessageType::IM_MESSAGE_GROUP_RESUME_LIKE << "," << ImMessageType::IM_MESSAGE_GROUP_JOB_INVITATION << ")";
  if (max_msg_id > 0) {
    os << " AND msg_id > " << max_msg_id;
  }
  os << " ORDER BY msg_id DESC LIMIT " << limit;
  mysqlpp::Query query(cp->query(os.str()));
  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while (mysqlpp::Row row = ret.fetch_row()) {
        msgid_res.insert((int64_t)row["msg_id"]);
      }
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::select_tb_group_message_index_n --> query_str:" << query.str() << " error:" <<
        e.what());
    return false;
  }

  LOG_INFO("MessageDBWrapper::select_tb_group_message_index_n --> ok. uid:" << uid << " group_id:" << group_id <<
      " max_msg_id:" << max_msg_id << " limit:" << limit << " msgid_res_count:" << msgid_res.size());
  return true;
}

bool MessageDBWrapper::update_tb_group_message_view_record(int64_t group_id, int64_t uid,
    int64_t last_read_msgid) {
  if (group_id <= 0) {
    return false;
  }
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::update_tb_group_message_view_record --> get db conn failed.");
    return false;
  }

  int64_t last_read_msgid_old = 0;
  std::ostringstream os;
  os << "SELECT last_read_msgid FROM tb_group_message_view_record " <<
      " WHERE group_id = " << group_id <<
      " AND receiver = " << uid <<
      " LIMIT 1";
  mysqlpp::Query query(cp->query(os.str()));
  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while (mysqlpp::Row row = ret.fetch_row()) {
        last_read_msgid_old = (int64_t)row["last_read_msgid"];
      }
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::update_tb_group_message_view_record --> query_str:" << query.str() << " error:" <<
        e.what());
  }

  bool needUpdate = false;
  query.reset();
  if (last_read_msgid_old <= 0) {
    query << "INSERT INTO tb_group_message_view_record " <<
      "(group_id, receiver, last_read_msgid) values (" <<
      group_id << "," <<
      uid << "," <<
      last_read_msgid << ")";
    needUpdate = true;
  } else if (last_read_msgid > last_read_msgid_old) {
    query << "UPDATE tb_group_message_view_record " <<
        " SET last_read_msgid = " << last_read_msgid <<
        " WHERE group_id = " << group_id <<
        " AND receiver = " << uid;
    needUpdate = true;
  }
  if (!needUpdate) {
    return true;
  }
  try {
    if (!query.exec()) {
      LOG_ERROR("MessageDBWrapper::update_tb_group_message_view_record --> query return false. query_str:" << query.str());
      return false;
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::update_tb_group_message_view_record --> query_str:" << query.str() << " error:" << e.what());
    return false;
  }

  LOG_INFO("MessageDBWrapper::update_tb_group_message_view_record --> query ok. group_id:" << group_id <<
      " receiver:" << uid << " last_read_msgid:" << last_read_msgid);
  return true;
}

bool MessageDBWrapper::select_tb_group_message_view_record(int64_t group_id, int64_t uid, int64_t& last_read_msgid) {
  mysqlpp::ScopedConnection cp(*conn_pool_);
  if (!cp) {
    LOG_ERROR("MessageDBWrapper::select_tb_group_message_view_record --> get db conn failed. group_id:" << group_id <<
        " receiver:" << uid);
    return false;
  }

  std::ostringstream os;
  os << "SELECT last_read_msgid FROM tb_group_message_view_record WHERE group_id = " << group_id <<
      " AND receiver = " << uid << " LIMIT 1";
  mysqlpp::Query query(cp->query(os.str()));
  try {
    mysqlpp::UseQueryResult ret = query.use();
    if (ret) {
      while (mysqlpp::Row row = ret.fetch_row()) {
        last_read_msgid = (int64_t)row["last_read_msgid"];
      }
    }
  } catch (mysqlpp::Exception &e) {
    LOG_ERROR("MessageDBWrapper::select_tb_group_message_view_record --> query_str:" << query.str() << " error:" <<
        e.what());
    return false;
  }

  LOG_INFO("MessageDBWrapper::select_tb_group_message_view_record --> query ok. group_id:" << group_id <<
      " receiver:" << uid << " last_read_msgid:" << last_read_msgid);
  return true;
}

}

